package assignment3;

import java.sql.*;
import java.util.*;
import java.util.Date;

public class Assignment {

	/***
	 * Returns a new Connection to the database.
	 * @param classForName - String, the class to use in order to open the connection.
	 * @param connectionURL - the url/i path to the database.
	 * @return the connection object.
	 * @throws ClassNotFoundException if the class is not found.
	 * @throws SQLException if connection to the database cannot be created.
	 * @param username - String, self-explanatory.
	 * @param password - String, self-explanatory.
	 */
	private static Connection createConnection(String classForName,String connectionURL, 
			String username, String password) throws ClassNotFoundException, 
			SQLException{
		Class.forName(classForName);

		if(username == null && password == null){
			return DriverManager.getConnection(connectionURL);
		}else{
			return DriverManager.getConnection(connectionURL, username, password);
		}
	}

	/***
	 * Returns a new Connection to the Cassandra database.
	 * @return the connection object.
	 * @throws ClassNotFoundException if the Cassandra jars are not found.
	 * @throws SQLException if connection to the database cannot be created.
	 */
	private static Connection createCassandraConnection() throws ClassNotFoundException, SQLException{
		return createConnection("org.apache.cassandra.cql.jdbc.CassandraDriver","jdbc:cassandra://cas1.ise.bgu.ac.il:9160/idanmora", null, null);
	}

	/***
	 * Returns a new Connection to the Oracle database.
	 * @return the connection object.
	 * @throws ClassNotFoundException if the ojdbc jar is not found.
	 * @throws SQLException if connection to the database cannot be created.
	 */
	private static Connection createOracleConnection() throws ClassNotFoundException, SQLException{
		return createConnection("oracle.jdbc.driver.OracleDriver", "jdbc:oracle:thin:@ora1.ise.bgu.ac.il:1521/ORACLE","idanmora", "idan1403");
	}

	/***
	 * This function validates that a sting is not null and empty.
	 * @param s - String, the string to validate.
	 * @return boolean value - true if is not null and not empty; false otherwise.
	 */
	private static boolean validateString(String s){
		return (s != null && !s.isEmpty());
	}

	/***
	 * The function create column family with the name defined by tableName and the 
	 * primary key name, of the type text, defined by keyName.
	 * @param tableName - String, the table name to be created.
	 * @param keyName - String, the key name of the table to be created.
	 */
	public static void createTable(String tableName, String keyName){
		if(validateString(tableName) && validateString(keyName)){

			Connection conn = null;
			Statement st= null;

			try{
				String data = "CREATE COLUMNFAMILY "+ tableName +" ("+ keyName
						+" text PRIMARY KEY);";
				conn = createCassandraConnection();
				st = conn.createStatement();
				st.executeUpdate(data);

			}catch(ClassNotFoundException e){
				e.printStackTrace();
			}catch(SQLException e){
				e.printStackTrace();
			}finally{
				if(st!=null){
					try {
						st.close();
					} catch (SQLException e) {
						e.printStackTrace();
					}
				}
				try{
					if(conn!=null){
						conn.close();
					}
				}catch (SQLException e) {
					e.printStackTrace();
				}
			}
		}
	}

	/***
	 * the function using createTable and creates the tables: MediaItems, USERS, 
	 * HistoryByUsers, HistoryByItems.
	 */
	public static void createTables(){
		createTable("MediaItems", "TITLE");
		createTable("USERS", "USERNAME");
		createTable("HistoryByUsers", "USERNAME");
		createTable("HistoryByItems", "TITLE");
	}

	/***
	 * The function copy all the information, except MID, from the Oracle 
	 * table MediaItems to the Cassandra column family MediaItems.
	 */
	public static void fillMediaItems(){
		Connection cassandraConn = null, OracleConn = null;
		ResultSet rs = null;
		Statement oracleSt = null;
		PreparedStatement cassandraPs = null;

		try{
			OracleConn = createOracleConnection();
			String getMediaItems = "SELECT TITLE, PROD_YEAR, TITLE_LENGTH FROM MediaItems";
			oracleSt = OracleConn.createStatement();
			rs = oracleSt.executeQuery(getMediaItems);
			cassandraConn = createCassandraConnection();
			String data = "INSERT INTO MediaItems (TITLE, PROD_YEAR, TITLE_LENGTH) "
					+ "VALUES (?,?,?);"; 
			cassandraPs = cassandraConn.prepareStatement(data);

			while(rs.next()){
				cassandraPs.setString(1, rs.getString("TITLE"));
				cassandraPs.setString(2, rs.getString("PROD_YEAR"));
				cassandraPs.setString(3, rs.getString("TITLE_LENGTH"));
				cassandraPs.execute();
			}
			rs.close();


		} catch (ClassNotFoundException e) {
			e.printStackTrace();
		} catch (SQLException e) {
			e.printStackTrace();
		}finally{
			try{
				if(oracleSt!=null){
					oracleSt.close();
				}
			} catch (SQLException e) {
				e.printStackTrace();
			}
			try{
				if(cassandraPs!=null){
					cassandraPs.close();
				}
			} catch (SQLException e) {
				e.printStackTrace();
			}
			try{
				if(rs != null){
					rs.close();
				}
			}
			catch (SQLException e) {
				e.printStackTrace();
			}
			try {
				if(cassandraConn != null){
					cassandraConn.close();
				}
			} catch (SQLException e) {
				e.printStackTrace();
			}
			try {
				if(OracleConn != null){
					OracleConn.close();
				}
			} catch (SQLException e) {
				e.printStackTrace();
			}
		}
	}

	/***
	 * The function returns true if the received username exist in the table 
	 * USERS otherwise false.
	 * @param username - String, the name of the user to check in the table USERS.  
	 * @return boolean value - true if the received username exist in the table
	 * 		   USERS; otherwise false.
	 */
	public static boolean isExistUsername(String username){
		boolean ans = false;
		if(validateString(username)){
			Connection conn = null;
			Statement st = null;
			ResultSet rs = null;

			try{
				conn = createCassandraConnection();
				String data = "SELECT * FROM USERS USING CONSISTENCY ONE";
				st = conn.createStatement();
				rs = st.executeQuery(data);
				while(rs.next()){
					if(rs.getString(rs.getMetaData().getColumnName(1)).equals(username)){
						ans = true;
						break;
					}

				}
				rs.close();
			}
			catch(SQLException e){
				e.printStackTrace();
			}
			catch (ClassNotFoundException e) {
				e.printStackTrace();
			}
			finally{
				try{
					if(st!=null){
						st.close();
					}
				}catch (SQLException e) {
					e.printStackTrace();
				}
				try{
					if(rs!=null){
						rs.close();
					}
				}catch (SQLException e) {
					e.printStackTrace();
				}
				try{
					if(conn!=null){
						conn.close();
					}
				}catch (SQLException e) {
					e.printStackTrace();
				}
			}
		}
		return ans;
	}

	/***
	 * The function checks if the username exist in the column family USERS using 
	 * consistency ONE, in case of positive answer return null, else insert the user 
	 * to the table USERS using consistency ALL and return the username.
	 * @param username - String, the username.
	 * @param password - String, the password.
	 * @param first_name - String, the first name of the user.
	 * @param last_name - String, the last name of the user.
	 * @param birth_date - String, the birth date in ms since 1970.
	 * @param registration_date - String, the registration date in ms since 1970.
	 * @return the username in case the insert was successful; null otherwise.
	 */
	public static String insertUser(String username, String password, String first_name,
			String last_name, String birth_date, String registration_date){
		String toReturn = null;
		//username is being validated in isExistUsername
		if(validateString(password) && validateString(first_name) && validateString(last_name)
				&& validateString(birth_date) && validateString(registration_date)){

			Connection conn = null;
			Statement st = null;
			if(!isExistUsername(username)){
				try {
					long birthDateLong = Long.valueOf(birth_date);
					long registrationDateLong = Long.valueOf(registration_date);


					if( birthDateLong >= 0  && registrationDateLong >= 0){
						conn = createCassandraConnection();

						String data = "INSERT INTO USERS(USERNAME,PASSWORD,FIRST_NAME,"
								+ "LAST_NAME,BIRTH_DAY,REGISTRATION_DATE) VALUES("+username
								+","+password+","+first_name+","+last_name+","+birth_date+
								","+registration_date+") USING CONSISTENCY ALL";

						st = conn.createStatement();
						st.executeUpdate(data);
						toReturn = username.toString();
					}

				} catch (ClassNotFoundException | SQLException e) {
					e.printStackTrace();

				}catch(NumberFormatException e){
					e.printStackTrace();
				}finally{
					if(st!=null){
						try {
							st.close();
						} catch (SQLException e) {
							e.printStackTrace();
						}
					}

					if(conn != null){
						try {
							conn.close();
						} catch (SQLException e) {
							e.printStackTrace();
						}
					}
				}
			}

		}

		return toReturn;
	}

	/***
	 * This function validates that given Integer is not null and is positive.
	 * @param integer - Integer, the number to validate.
	 * @return boolean value - true if the integer is not null and positive; false otherwise.
	 */
	private static boolean validateInteger(Integer integer) {
		return ((integer != null) && (integer > 0));
	}

	/***
	 * retrieves from the table MediaItems first top_n items.
	 * @param top_n - Integer, the top number
	 * @return List of objects MediaItems retrieves from the table MediaItems, 
	 * 		   first top_n items.
	 */
	public static List<Mediaitems> getTopNItems(Integer top_n){
		Vector<Mediaitems> items = new Vector<Mediaitems>();

		if(validateInteger(top_n)){
			Connection con = null;
			Statement st = null;
			ResultSet rs = null;
			try {
				con = createCassandraConnection();
				String data = "SELECT * FROM MediaItems";
				st = con.createStatement();
				rs = st.executeQuery(data);

				int i = 0;
				while(rs.next() && i < top_n){
					Mediaitems item = new Mediaitems(rs.getString("TITLE"), 
							rs.getShort("PROD_YEAR"), rs.getShort("TITLE_LENGTH"));
					items.add(item);
					i++;
				}

				rs.close();
			} catch (ClassNotFoundException | SQLException e) {
				e.printStackTrace();
			}finally{
				if(rs != null){
					try {
						rs.close();
					} catch (SQLException e) {
						e.printStackTrace();
					}
				}
				if(st != null){
					try {
						st.close();
					} catch (SQLException e) {
						e.printStackTrace();
					}
				}
				if(con != null){
					try {
						con.close();
					} catch (SQLException e) {
						e.printStackTrace();
					}
				}
			}
		}

		return items;
	}

	/***
	 * The function compares received values with existing in the data base.
	 * @param username - String, the user name needed to be checked
	 * @param password - String, the user's password.
	 * @return USERNAME if the values are equal to the values in the table; otherwise NULL.
	 */
	public static String validateUser(String username, String password){
		String ans = null;
		if(validateString(username) && validateString(password)){
			Connection con = null;
			Statement st = null;
			ResultSet rs = null;

			String data = "SELECT * FROM USERS USING CONSISTENCY ONE";
			try {
				con = createCassandraConnection();
				st = con.createStatement();
				rs = st.executeQuery(data);

				while(rs.next()){
					String userFromTable = rs.getString("USERNAME");
					String passwordFromTable = rs.getString("PASSWORD");
					if(userFromTable != null && userFromTable.equals(username) 
							&& passwordFromTable != null && passwordFromTable.equals(password)){
						ans = userFromTable.toString();
						break;
					}
				}
				rs.close();
			} catch (ClassNotFoundException | SQLException e) {
				e.printStackTrace();
			}finally{
				if(rs != null){
					try {
						rs.close();
					} catch (SQLException e) {
						e.printStackTrace();
					}
				}
				if(st != null){
					try {
						st.close();
					} catch (SQLException e) {
						e.printStackTrace();
					}
				}
				if(con != null){
					try {
						con.close();
					} catch (SQLException e) {
						e.printStackTrace();
					}
				}
			}
		}
		return ans;
	}

	/***
	 * The function receives two Strings (username and title) .
	 * The function inserts with consistency ANY a column to the column 
	 * family HistoryByUsers associated with the key username. The column 
	 * name should be current time in ms since 1970; the column value is the title.
	 * The function inserts with the consistency ANY a column to the column family 
	 * HistoryByItems associated with the key title. The column name should be 
	 * current time in ms since 1970; the column value is the username.
	 * @param username - String, the user name.
	 * @param title - String, the password.
	 */
	public static void insertToHistory(String username, String title){
		if(validateString(username) && validateString(title)){
			String InsertHistoryByUsers = "INSERT INTO HistoryByUsers(USERNAME,"+
					System.currentTimeMillis()+") VALUES('"+username+"','"+
					title+"') USING CONSISTENCY ANY";
			String insertHistoryByItems = "INSERT INTO HistoryByItems(TITLE,"+
					System.currentTimeMillis()+") VALUES('"+title+"','"+
					username+"') USING CONSISTENCY ANY";
			Connection con = null;
			Statement st = null;

			try {
				con = createCassandraConnection();
				st = con.createStatement();
				st.executeUpdate(InsertHistoryByUsers);
				st.executeUpdate(insertHistoryByItems);

			} catch (ClassNotFoundException | SQLException e) {
				e.printStackTrace();
			}finally{
				if(st != null){
					try {
						st.close();
					} catch (SQLException e) {
						e.printStackTrace();
					}
				}
				if(con != null){
					try {
						con.close();
					} catch (SQLException e) {
						e.printStackTrace();
					}
				}
			}
		}

	}

	/**
	 * The function receives Strings (username) 
	 * The function retrieves with consistency ONE from the column family HistoryByUser item's users
	 * The function return List of pairs <title,viewtime> sorted by VIEWTIME in descending order
	 * @param username - String, the username.
	 * @return List of pairs < title,viewtime> sorted by VIEWTIME in descending order
	 */
	public static List<Pair> getHistoryByUser(String username){
		Vector<Pair> ans = new Vector<Pair>();

		if(validateString(username)){
			String data = "SELECT * FROM HistoryByUsers USING CONSISTENCY ONE where USERNAME = '"+username+"';";
			pupolateList(ans, data);
		}

		return ans;
	}

	/**
	 * The function receives Strings (title) 
	 * The function retrieves with consistency ONE from the column family HistoryByItems item's users
	 * The function return List of pairs < username,viewtime> sorted by VIEWTIME in descending order
	 * @param title - String, the item title.
	 * @return List of pairs < username,viewtime> sorted by VIEWTIME in descending order
	 */
	public static List<Pair> getHistoryByItem(String title){
		Vector<Pair> ans = new Vector<Pair>();

		if(validateString(title)){
			String data = "SELECT * FROM HistoryByItems USING CONSISTENCY ONE where TITLE = '"+title+"';";
			pupolateList(ans, data);
		}


		return ans;

	}

	/**
	 * the function populate the List<Pair> with a given result from the cql data string 
	 * sorted and reversed it by viewTime
	 * @param ans- the given list
	 * @param data- cql string
	 */
	private static void pupolateList(List<Pair> ans, String data) {
		Connection con = null;
		Statement st = null;
		ResultSet rs = null;

		try {
			con = createCassandraConnection();
			st = con.createStatement();
			rs = st.executeQuery(data);

			while(rs.next()){
				//skipping the key column
				for(int i =2; i<rs.getMetaData().getColumnCount()+1; i++){
					Pair p = new Pair();
					p.setKey(rs.getString(rs.getMetaData().getColumnName(i)));
					p.setViewtime(new Date(Long.valueOf(rs.getMetaData().getColumnName(i))));
					ans.add(p);
				}

			}
			rs.close();
			Collections.sort(ans);
			Collections.reverse(ans);

		} catch (ClassNotFoundException | SQLException | NumberFormatException e) {
			e.printStackTrace();
		}finally{
			if(rs != null)
				try {
					rs.close();
				} catch (SQLException e) {
					e.printStackTrace();
				}
			if(st != null){
				try {
					st.close();
				} catch (SQLException e) {
					e.printStackTrace();
				}
			}
			if(con != null){
				try {
					con.close();
				} catch (SQLException e) {
					e.printStackTrace();
				}
			}

		}
	}

	/***
	 * The function receives an integer number n.
	 * The function retrieves with consistency ONE from the column family Users number of registered users in the past n days.
	 * The function return integer number.
	 * @param n - Integer, the number of days.
	 * @return Integer number of users.
	 */
	public static Integer getNumberOfRegistredUsers(Integer n){
		Integer ans =null;

		if(validateInteger(n)){
			String data = "SELECT REGISTRATION_DATE FROM USERS USING CONSISTENCY ONE;";
			Statement st = null;
			Connection con = null;
			ResultSet rs = null;
			
			//the GregorianCalendar Object is initiated with the current time of client
			GregorianCalendar calendar = new GregorianCalendar();

			//subtracts the amounts of days 
			calendar.add(GregorianCalendar.DAY_OF_MONTH, -1*n-1);
			
			GregorianCalendar current = new GregorianCalendar();
			current.add(GregorianCalendar.DAY_OF_MONTH, 1);

			try {
				con = createCassandraConnection();
				st = con.createStatement();
				rs = st.executeQuery(data);
				
				ans = 0;
				while(rs.next()){
					String registration = rs.getString(rs.getMetaData().getColumnName(1));
					Date registrationDate = new Date(Long.valueOf(registration));
					
					if(registrationDate.after(calendar.getTime()) && 
							registrationDate.before(current.getTime())){
						ans++;
					}
				}
				
			} catch (ClassNotFoundException | SQLException | NumberFormatException e) {
				e.printStackTrace();
			}
		}


		return ans;
	}
	
	/***
	 * The function retrieves with consistency ONE from the column family Users all users
	 * @return List of objects Users.
	 */
	public static List<Users> getUsers(){
		Vector<Users> ans = null;
		
		String data = "SELECT * FROM USERS USING CONSISTENCY ONE;";
		Statement st = null;
		Connection con = null;
		ResultSet rs = null;
		
		try {
			con = createCassandraConnection();
			st = con.createStatement();
			rs = st.executeQuery(data);
			ans = new Vector<Users>();
			while(rs.next()){
				Users user = new Users();
				user.setUsername(rs.getString("USERNAME"));
				user.setPassword(rs.getString("PASSWORD"));
				user.setFirstName(rs.getString("FIRST_NAME"));
				user.setLastName(rs.getString("LAST_NAME"));
				user.setDateOfBirth(new Date(Long.valueOf(rs.getString("BIRTH_DAY"))));
				user.setRegistrationDate(new Date(Long.valueOf(rs.getString("REGISTRATION_DATE"))));
				ans.add(user);
			}
		} catch (ClassNotFoundException | SQLException | NumberFormatException e) {
			e.printStackTrace();
		}
		
		return ans;
	}
	
	/***
	 * The function receives  String (title) .
	 * Using previous functions the function retrieves user's information that exist in the history of the given item
	 * @param title - String, the title of the item.
	 * @return List of objects Users.
	 */
	public static List<Users> getUsersByItemHistory(String title) {
		Vector<Users> ans = null;
		
		if(validateString(title)){
			List<Pair> usersFromItemHistory = getHistoryByItem(title);
			List<Users> users = getUsers();
			
			ans = new Vector<Users>();
			for(Pair p: usersFromItemHistory){
				String username = p.getKey();
				for(Users u: users){
					if(!ans.contains(u) && u.getUsername().equals(username)){
						ans.add(u);
						break;
					}
				}
			}
		}
		
		return ans;
	}
	
	/***
	 * Using the column family HistoryByItems with consistency ONE the function calculates the similarity score using Jaccard similarity.
	 * @param title1 - String, the first item title.
	 * @param title2 - String, the second item title.
	 * @return double value - the Jaccard similarity.
	 */
	public static Double getItemsSimilarity(String title1,String title2) {
		Double ans = null;
		if(validateString(title1) && validateString(title2)){
			List<Users> listTitle1 = getUsersByItemHistory(title1);
			List<Users> listTitle2 = getUsersByItemHistory(title2);
			Vector<Users> numerator = new Vector<Users>();
			HashSet<Users> denominator = new HashSet<Users>(listTitle1);
			
			for(Users u:listTitle1){
				if(listTitle2.contains(u)){
					numerator.add(u);
				}
			}
			
			denominator.addAll(listTitle2);
			
			if(denominator.size() > 0){
				ans = (double)numerator.size()/denominator.size();
			}
		}
		return ans;
		
	}

}
